package analyse;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.math3.stat.descriptive.DescriptiveStatistics;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;

import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.UUID;

/**
 * 使用Stream实现解析器 采样间隔通过流操作自身的窗口功能实现 计算平均响应时延是采用commons中的DescriptiveStatistics实现
 *
 * @author pf
 */
public class SamplerAnalyseStreamNew implements Runnable {

	private String brokerUrl;
	private String topic;
	private int interval;
	public SamplerAnalyseStreamNew(String brokerUrl, String topic, int interval) {

		this.brokerUrl = brokerUrl;
		this.topic = topic;
		this.interval = interval;
	}

	public static void main(String[] args) {

		new Thread(new SamplerAnalyseStreamNew("172.18.84.143:9092", "test", 10)).start();
	}

	/**
	 * @param brokerUrl kafka连接地址
	 * @param topic     topic名称
	 * @param interval  采样间隔，单位：秒
	 */
	@Override
	public void run() {

		Properties props = new Properties();

		props.put(StreamsConfig.APPLICATION_ID_CONFIG, UUID.randomUUID().toString());
		props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
		props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
		props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

		StreamsBuilder builder = new StreamsBuilder();

		KStream<String, String> textLines = builder.stream(topic);

		textLines.process(() -> new MyNewProcess(interval));

		KafkaStreams streams = new KafkaStreams(builder.build(), props);
		streams.start();
	}
}

class MyNewProcess implements Processor<String, String> {

	private final static double LEVEL_1_PERCENT = 90;
	private final static double LEVEL_2_PERCENT = 95;
	private final static double LEVEL_3_PERCENT = 99;
	private ProcessorContext context;
	private int interval;
	private HashMap<String, ArrayList<JSONObject>> samplers = new HashMap<String, ArrayList<JSONObject>>();

	public MyNewProcess(int interval) {

		this.interval = interval;
	}

	@Override
	public void init(ProcessorContext context) {

		this.context = context;
		this.context.schedule(Duration.ofSeconds(interval), PunctuationType.STREAM_TIME, (timeStamp) -> {
			long start = System.currentTimeMillis();
			JSONArray res = analyse(timeStamp);
			System.out.println((System.currentTimeMillis() - start) + res.toJSONString());
		});
	}

	@Override
	public void process(String key, String value) {

//		System.out.println(value);

		JSONObject valueJson = JSONObject.parseObject(value);

		ArrayList<JSONObject> tempList = samplers.get(key);
		if (tempList == null) {
			tempList = new ArrayList<JSONObject>();
			samplers.put(key, tempList);
		}

		tempList.add(valueJson);
	}

	@Override
	public void close() {
	}

	public JSONArray analyse(long timeStamp) {

		JSONArray resArry = new JSONArray();

		JSONObject total = new JSONObject();

		resArry.add(total);

		int totalSuccessCount = 0;
		int totalFailCount = 0;
		int totalThreads = 0;
		HashMap<String, Integer> totalThreadsMap = new HashMap<String, Integer>();
		DescriptiveStatistics totalDelayStatistics = new DescriptiveStatistics();
		long totalSendBytes = 0L;
		long totalReceiveBytes = 0L;

		// 根据不同的SampleLabel遍历
		for (Entry<String, ArrayList<JSONObject>> entry : samplers.entrySet()) {

			String key = entry.getKey();
			ArrayList<JSONObject> value = entry.getValue();

			int tempSuccessCount = 0;
			int tempFailCount = 0;
			int tempThreads = 0;
			HashMap<String, Integer> tempThreadsMap = new HashMap<String, Integer>();
			DescriptiveStatistics tempDelayStatistics = new DescriptiveStatistics();
			long tempSendBytes = 0L;
			long tempReceiveBytes = 0L;

			for (JSONObject jsonObject : value) {

				String currentHost = jsonObject.getString("currentHost");
				String threadGroupName = jsonObject.getString("threadGroupName");

				tempThreadsMap.put(currentHost, jsonObject.getInteger("threadCount"));
				totalThreadsMap.put(currentHost + threadGroupName.substring(0, threadGroupName.indexOf("-")),
						jsonObject.getInteger("threadCount"));

				if (jsonObject.getBooleanValue("success")) {
					tempSuccessCount = tempSuccessCount + jsonObject.getIntValue("sampleCount");
				} else {
					tempFailCount = tempFailCount + jsonObject.getIntValue("sampleCount");
				}
				tempDelayStatistics.addValue(jsonObject.getDoubleValue("delayMS"));
				totalDelayStatistics.addValue(jsonObject.getDoubleValue("delayMS"));
				tempSendBytes = tempSendBytes + jsonObject.getLongValue("sentBytes");
				tempReceiveBytes = tempReceiveBytes + jsonObject.getLongValue("receiveBytes");
			}

			// 计算当前采样的并发数
			for (Entry<String, Integer> tempEntry : tempThreadsMap.entrySet()) {
				tempThreads = tempThreads + tempEntry.getValue();
			}

			JSONObject temp = new JSONObject();
			resArry.add(temp);

			temp.put("timeStamp", timeStamp);
			temp.put("label", key);
			temp.put("successCount", tempSuccessCount);
			temp.put("failCount", tempFailCount);
			temp.put("qps", (tempSuccessCount + tempFailCount) / interval);
			temp.put("threads", tempThreads);
			temp.put("sendBytes/s", tempSendBytes / interval);
			temp.put("receiveBytes/s", tempReceiveBytes / interval);
			temp.put("level_1_percent", tempDelayStatistics.getPercentile(LEVEL_1_PERCENT));
			temp.put("level_2_percent", tempDelayStatistics.getPercentile(LEVEL_2_PERCENT));
			temp.put("level_3_percent", tempDelayStatistics.getPercentile(LEVEL_3_PERCENT));
			temp.put("maxDelay", tempDelayStatistics.getMax());
			temp.put("minDelay", tempDelayStatistics.getMin());

			totalSuccessCount = totalSuccessCount + tempSuccessCount;
			totalFailCount = totalFailCount + tempFailCount;
			totalSendBytes = totalSendBytes + tempSendBytes;
			totalReceiveBytes = totalReceiveBytes + tempReceiveBytes;
		}

		// 计算总的并发数
		for (Entry<String, Integer> tempEntry : totalThreadsMap.entrySet()) {
			totalThreads = totalThreads + tempEntry.getValue();
		}

		total.put("timeStamp", timeStamp);
		total.put("label", "total");
		total.put("successCount", totalSuccessCount);
		total.put("failCount", totalFailCount);
		total.put("qps", (totalSuccessCount + totalFailCount) / interval);
		total.put("threads", totalThreads);
		total.put("sendBytes/s", totalSendBytes / interval);
		total.put("receiveBytes/s", totalReceiveBytes / interval);
		total.put("level_1_percent", totalDelayStatistics.getPercentile(LEVEL_1_PERCENT));
		total.put("level_2_percent", totalDelayStatistics.getPercentile(LEVEL_2_PERCENT));
		total.put("level_3_percent", totalDelayStatistics.getPercentile(LEVEL_3_PERCENT));
		total.put("maxDelay", totalDelayStatistics.getMax());
		total.put("minDelay", totalDelayStatistics.getMin());

		samplers.clear();

		return resArry;

	}

}
